package com.atguigu.realtime.app.dwm;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.atguigu.realtime.app.BaseAppV2;
import com.atguigu.realtime.bean.OrderDetail;
import com.atguigu.realtime.bean.OrderInfo;
import com.atguigu.realtime.bean.OrderWide;
import com.atguigu.realtime.commont.Constant;
import com.atguigu.realtime.util.DimUtil;
import com.atguigu.realtime.util.JdbcUtil;
import com.atguigu.realtime.util.RedisUtil;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.RichMapFunction;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.co.ProcessJoinFunction;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.util.Collector;
import redis.clients.jedis.Jedis;

import java.sql.Connection;
import java.time.Duration;
import java.util.Map;

import static com.atguigu.realtime.commont.Constant.TOPIC_DWD_ORDER_DETAIL;
import static com.atguigu.realtime.commont.Constant.TOPIC_DWD_ORDER_INFO;

/**
 * @Author lizhenchao@atguigu.cn
 * @Date 2021/11/26 11:10
 */
public class DwmOrderWideApp_Cache extends BaseAppV2 {
    public static void main(String[] args) {
        new DwmOrderWideApp_Cache().init(3003, 1, "DwmOrderWideApp_Cache", "DwmOrderWideApp_Cache",
                                         TOPIC_DWD_ORDER_INFO, TOPIC_DWD_ORDER_DETAIL);
    }
    
    @Override
    protected void run(StreamExecutionEnvironment env,
                       Map<String, DataStreamSource<String>> streamMap) {
        // 1. 双流join  事实表join
        SingleOutputStreamOperator<OrderWide> orderWideStreamWithoutDims = factJoin(streamMap);
        // 2. join维度信息
        SingleOutputStreamOperator<OrderWide> orderWideStreamWithDims = dimJoin(orderWideStreamWithoutDims);
        orderWideStreamWithDims.print();
        // 3. 把数据写入到Kafka中
        
    }
    
    private SingleOutputStreamOperator<OrderWide> dimJoin(SingleOutputStreamOperator<OrderWide> stream) {
        /*
        针对每个OrderWide去连接Phoenix, 读取对应的维度数据
        根据各个id
         */
        return stream
            .map(new RichMapFunction<OrderWide, OrderWide>() {
    
                private Jedis redisClient;
                private Connection conn;
                
                @Override
                public void open(Configuration parameters) throws Exception {
                    conn = JdbcUtil.getPhoenixConnection(Constant.PHOENIX_URL);
    
                    redisClient = RedisUtil.getRedisClient();
    
                }
                
                @Override
                public OrderWide map(OrderWide orderWide) throws Exception {
                    // 补充维度信息
                    // 1. 补充user_info的信息
                    JSONObject userInfo = DimUtil.readDim(conn, redisClient, "dim_user_info", orderWide.getUser_id());
                    orderWide.setUser_gender(userInfo.getString("GENDER"));
                    orderWide.calcUserAge(userInfo.getString("BIRTHDAY"));
                    // 2. 补充省份信息
                    JSONObject baseProvince = DimUtil.readDim(conn, redisClient, "dim_base_province", orderWide.getProvince_id());
                    orderWide.setProvince_3166_2_code(baseProvince.getString("ISO_3166_2"));
                    orderWide.setProvince_area_code(baseProvince.getString("AREA_CODE"));
                    orderWide.setProvince_iso_code(baseProvince.getString("ISO_CODE"));
                    orderWide.setProvince_name(baseProvince.getString("NAME"));
                    
                    // 3. 补充sku
                    JSONObject skuInfo = DimUtil.readDim(conn, redisClient, "dim_sku_info", orderWide.getSku_id());
                    orderWide.setSku_name(skuInfo.getString("SKU_NAME"));
                    orderWide.setOrder_price(skuInfo.getBigDecimal("PRICE"));
                    
                    // 给spu_id tm_id category3_id进行赋值, 因为他们只有知道了sku才能获取到这三个id
                    orderWide.setSpu_id(skuInfo.getLong("SPU_ID"));
                    orderWide.setTm_id(skuInfo.getLong("TM_ID"));
                    orderWide.setCategory3_id(skuInfo.getLong("CATEGORY3_ID"));
                    
                    // 4. 补充spu
                    JSONObject spuInfo = DimUtil.readDim(conn, redisClient, "dim_spu_info", orderWide.getSpu_id());
                    orderWide.setSpu_name(spuInfo.getString("SPU_NAME"));
                    
                    
                    // 5. 补充tm
                    JSONObject tmInfo = DimUtil.readDim(conn, redisClient, "dim_base_trademark", orderWide.getTm_id());
                    orderWide.setTm_name(tmInfo.getString("TM_NAME"));
                    
                    
                    // 6. 补充c3
                    JSONObject c3Info = DimUtil.readDim(conn, redisClient, "dim_base_category3", orderWide.getCategory3_id());
                    orderWide.setCategory3_name(c3Info.getString("NAME"));
                    
                    
                    return orderWide;
                }
            });
        
    }
    
    private SingleOutputStreamOperator<OrderWide> factJoin(Map<String, DataStreamSource<String>> streamMap) {
        KeyedStream<OrderInfo, Long> orderInfoStream = streamMap
            .get(TOPIC_DWD_ORDER_INFO)
            .map(json -> JSON.parseObject(json, OrderInfo.class))
            .assignTimestampsAndWatermarks(
                WatermarkStrategy
                    .<OrderInfo>forBoundedOutOfOrderness(Duration.ofSeconds(3))
                    .withTimestampAssigner((info, ts) -> info.getCreate_ts())
            )
            .keyBy(OrderInfo::getId);
        KeyedStream<OrderDetail, Long> orderDetailStream = streamMap
            .get(TOPIC_DWD_ORDER_DETAIL)
            .map(json -> JSON.parseObject(json, OrderDetail.class))
            .assignTimestampsAndWatermarks(
                WatermarkStrategy
                    .<OrderDetail>forBoundedOutOfOrderness(Duration.ofSeconds(3))
                    .withTimestampAssigner((detail, ts) -> detail.getCreate_ts())
            )
            .keyBy(OrderDetail::getOrder_id);
        
        return orderInfoStream
            .intervalJoin(orderDetailStream)
            .between(Time.seconds(-5), Time.seconds(5))
            .process(new ProcessJoinFunction<OrderInfo, OrderDetail, OrderWide>() {
                @Override
                public void processElement(OrderInfo left,
                                           OrderDetail right,
                                           Context ctx, Collector<OrderWide> out) throws Exception {
                    out.collect(new OrderWide(left, right));
                    
                }
            });
        
    }
}
/*
缓存: 内存

内置
    状态
外置
    redis-
  
    
---
状态:
    优点: 本地内存, 速度快.  flink管理, 使用方便
    缺点:  1. 占用flink的内存, 影响flink计算的的时候内存使用
           2. 维度发生变化, 怎么办?
                无法感知到维度的变化
                
redis:  旁路缓存
    优点: 1. 专门缓存服务器, 内存充足
          2. 解决维度变化的问题:
               dwd 动态分流可以感知到维度变化, 当维度变化的时候可以去redis更新(删除)对应的维度
               
     缺点:
        通过网络访问

缓存雪崩  缓存击穿  缓存穿透

 */